/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2022 All Rights Reserved.
 */
package com.example.spider.proxy.core.repository;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.file.FileReader;
import cn.hutool.core.util.CharsetUtil;
import com.example.spider.proxy.core.ProxyInfo;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.shoulder.core.util.JsonUtils;
import org.springframework.scheduling.annotation.Scheduled;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * 定期与本地文件同步
 * 适合总量比较小的，如小于1w个，否则占用内存较多
 *
 * @author lym
 */
@Slf4j
public class FileBasedMemoryProxyRepository extends AbstractProxyRepository {

    private final String storagePath;

    /**
     * 爬虫直接爬下来的原始池，不建议直接使用，失败率高
     */
    private ConcurrentHashMap<String, ProxyInfo> proxyInfoMap;

    private final LongAdder memoryVersion = new LongAdder();

    private volatile long fileVersion = 0;

    private ReentrantLock flushLock = new ReentrantLock();

    public FileBasedMemoryProxyRepository(String storagePath) {
        this.storagePath = storagePath;
    }

    public static void writeProxyIpToFile(Collection<? extends ProxyInfo> proxyInfoMap, String fileName) {
        // json存储，可压缩比比较高，先压缩
        List<String> allIp = proxyInfoMap.stream()
                .map(JsonUtils::toJson)
                .collect(Collectors.toList());
        FileUtil.touch(fileName);
        FileUtil.writeLines(allIp, fileName, CharsetUtil.UTF_8);
    }

    /**
     * 形式
     * {ProxyInfoJson1}
     * {ProxyInfoJson2}
     */
    public static Map<String, ProxyInfo> loadProxyIpsFromJsonsFile(String fileName) {
        if (!FileUtil.exist(fileName)) {
            return Collections.emptyMap();
        }
        FileReader fileReader = new FileReader(fileName);
        List<String> proxyIpJsonList = fileReader.readLines();
        return proxyIpJsonList.stream().parallel()
                .map(json -> JsonUtils.parseObject(json, ProxyInfo.class))
                .collect(Collectors.toMap(AbstractProxyRepository::calculateKey, p -> p, (p1, p2) -> p2));
    }

    @PostConstruct
    public void initExistProxyIno() {
        proxyInfoMap = new ConcurrentHashMap<>(loadProxyIpsFromJsonsFile(storagePath));
    }

    //@Async
    @Scheduled(cron = "*/10 * * * * ?")
    public void scheduledWriteCacheToFile() {
        //processMimvpPort();
        flushLock.lock();
        try {
            long currentMemoryVersion = memoryVersion.longValue();
            boolean hasChange = currentMemoryVersion != fileVersion;
            if (!hasChange) {
                return;
            }
            writeProxyIpToFile(proxyInfoMap.values(), storagePath);
            log.info("scheduled save memory ipProxy to file(size={}).", proxyInfoMap.size());
            fileVersion = currentMemoryVersion;
        } finally {
            flushLock.unlock();
        }
    }

    // ===============

    @Override
    public Iterator<ProxyInfo> getIterator() {
        return proxyInfoMap.values().iterator();
    }

    public static void appendProxyIpToFile(ProxyInfo proxyInfo, String filePath) {
        File file = FileUtil.touch(filePath);
        FileUtil.appendString(JsonUtils.toJson(proxyInfo), file, StandardCharsets.UTF_8);
    }

    @Override
    protected int size() {
        return proxyInfoMap.size();
    }

    @Override
    protected void doAdd(ProxyInfo proxyInfo) {
        // 合并来源
        proxyInfoMap.put(calculateKey(proxyInfo), proxyInfo);
        memoryVersion.increment();
    }

    @Override
    public ProxyInfo remove(ProxyInfo proxyInfo) {
        ProxyInfo r = proxyInfoMap.remove(calculateKey(proxyInfo));
        memoryVersion.increment();
        return r;
    }

}